

#include "postgres.h"

#include "executor/execParallel.h"
#include "executor/executor.h"
#include "executor/nodeAgg.h"
#include "executor/nodeAppend.h"
#include "executor/nodeBitmapHeapscan.h"
#include "executor/nodeCustom.h"
#include "executor/nodeForeignscan.h"
#include "executor/nodeHash.h"
#include "executor/nodeHashjoin.h"
#include "executor/nodeIncrementalSort.h"
#include "executor/nodeIndexonlyscan.h"
#include "executor/nodeIndexscan.h"
#include "executor/nodeMemoize.h"
#include "executor/nodeSeqscan.h"
#include "executor/nodeSort.h"
#include "executor/nodeSubplan.h"
#include "executor/tqueue.h"
#include "jit/jit.h"
#include "nodes/nodeFuncs.h"
#include "pgstat.h"
#include "storage/spin.h"
#include "tcop/tcopprot.h"
#include "utils/datum.h"
#include "utils/dsa.h"
#include "utils/lsyscache.h"
#include "utils/memutils.h"
#include "utils/snapmgr.h"

/*
 * 用于并行执行器通信的魔法数字。我们在这里使用大于任何 32 位整数的常量，以便值 < 2^32 能被单独的并行节点用来存储它们自己的状态。
 */
#define PARALLEL_KEY_EXECUTOR_FIXED		UINT64CONST(0xE000000000000001)
#define PARALLEL_KEY_PLANNEDSTMT		UINT64CONST(0xE000000000000002)
#define PARALLEL_KEY_PARAMLISTINFO		UINT64CONST(0xE000000000000003)
#define PARALLEL_KEY_BUFFER_USAGE		UINT64CONST(0xE000000000000004)
#define PARALLEL_KEY_TUPLE_QUEUE		UINT64CONST(0xE000000000000005)
#define PARALLEL_KEY_INSTRUMENTATION	UINT64CONST(0xE000000000000006)
#define PARALLEL_KEY_DSA				UINT64CONST(0xE000000000000007)
#define PARALLEL_KEY_QUERY_TEXT		UINT64CONST(0xE000000000000008)
#define PARALLEL_KEY_JIT_INSTRUMENTATION UINT64CONST(0xE000000000000009)
#define PARALLEL_KEY_WAL_USAGE			UINT64CONST(0xE00000000000000A)

#define PARALLEL_TUPLE_QUEUE_SIZE		65536

/*
 * 需要传递给并行工作者的固定大小随机数据。
 */
typedef struct FixedParallelExecutorState
{
	int64		tuples_needed;	/* 元组边界，见 ExecSetTupleBound */
	dsa_pointer param_exec;
	int			eflags;
	int			jit_flags;
} FixedParallelExecutorState;

/*
 * 用于累积每 PlanState 仪器的 DSM 结构。
 *
 * instrument_options：与 instrument.c 中的含义相同。
 *
 * instrument_offset：相对于该结构开始的偏移量，第一个 Instrumentation 对象的位置。这将取决于 plan_node_id 数组的长度。
 *
 * num_workers：工作者的数量。
 *
 * num_plan_nodes：计划节点的数量。
 *
 * plan_node_id：我们从并行工作者那里收集仪器的计划节点数组。该数组的长度由 num_plan_nodes 确定。
 */
struct SharedExecutorInstrumentation
{
	int			instrument_options;
	int			instrument_offset;
	int			num_workers;
	int			num_plan_nodes;
	int			plan_node_id[FLEXIBLE_ARRAY_MEMBER];
	/* num_plan_nodes * num_workers 个 Instrumentation 对象的数组随之而来 */
};
#define GetInstrumentationArray(sei) \
	(AssertVariableIsOfTypeMacro(sei, SharedExecutorInstrumentation *), \
	 (Instrumentation *) (((char *) sei) + sei->instrument_offset))

/* ExecParallelEstimate 的上下文对象。 */
typedef struct ExecParallelEstimateContext
{
	ParallelContext *pcxt;
	int			nnodes;
} ExecParallelEstimateContext;

/* ExecParallelInitializeDSM 的上下文对象。 */
typedef struct ExecParallelInitializeDSMContext
{
	ParallelContext *pcxt;
	SharedExecutorInstrumentation *instrumentation;
	int			nnodes;
} ExecParallelInitializeDSMContext;

/* 在并行领导者中运行的辅助函数。 */
static char *fc_ExecSerializePlan(Plan *fc_plan, EState *fc_estate);
static bool fc_ExecParallelEstimate(PlanState *fc_node,
								 ExecParallelEstimateContext *fc_e);
static bool fc_ExecParallelInitializeDSM(PlanState *fc_node,
									  ExecParallelInitializeDSMContext *fc_d);
static shm_mq_handle **fc_ExecParallelSetupTupleQueues(ParallelContext *fc_pcxt,
													bool fc_reinitialize);
static bool fc_ExecParallelReInitializeDSM(PlanState *fc_planstate,
										ParallelContext *fc_pcxt);
static bool fc_ExecParallelRetrieveInstrumentation(PlanState *fc_planstate,
												SharedExecutorInstrumentation *fc_instrumentation);

/* 在并行工作者中运行的辅助函数。 */
static DestReceiver *fc_ExecParallelGetReceiver(dsm_segment *fc_seg, shm_toc *fc_toc);

/*
 * 创建一个计划的序列化表示，以便发送给每个工作者。
 */
static char * fc_ExecSerializePlan(Plan *fc_plan, EState *fc_estate)
{
	PlannedStmt *fc_pstmt;
	ListCell   *fc_lc;

	/* 我们无法在原始计划上进行涂改，因此制作一个副本。 */
	fc_plan = copyObject(fc_plan);

	/*
	 * 工作者将启动自己副本的执行器，如果顶部节点有任何 resjunk 条目，该副本将插入一个垃圾过滤器。我们不想让这种情况发生，因为虽然 resjunk 列不应该返回给用户，但在这里，元组是返回给另一个后端的，该后端可能非常需要它们。因此，适当地改变目标列表。这算是一种hack；可能有更好的方法来做到这一点……
	 */
	foreach(fc_lc, fc_plan->targetlist)
	{
		TargetEntry *fc_tle = lfirst_node(TargetEntry, fc_lc);

		fc_tle->resjunk = false;
	}

	/*
	 * 创建一个虚拟 PlannedStmt。出于我们的目的，大多数字段不需要有效，但工作者至少需要一个最小的 PlannedStmt 来启动执行器。
	 */
	fc_pstmt = makeNode(PlannedStmt);
	fc_pstmt->commandType = CMD_SELECT;
	fc_pstmt->queryId = pgstat_get_my_query_id();
	fc_pstmt->hasReturning = false;
	fc_pstmt->hasModifyingCTE = false;
	fc_pstmt->canSetTag = true;
	fc_pstmt->transientPlan = false;
	fc_pstmt->dependsOnRole = false;
	fc_pstmt->parallelModeNeeded = false;
	fc_pstmt->planTree = fc_plan;
	fc_pstmt->rtable = fc_estate->es_range_table;
	fc_pstmt->resultRelations = NIL;
	fc_pstmt->appendRelations = NIL;

	/*
	 * 仅传输并行安全的子计划，在列表中为不安全的子计划留一个 NULL "孔"（以保留安全子计划的列表索引）。这确保工作者不会尝试运行，甚至不会对不安全的子计划进行 ExecInitNode。这一点很重要，例如，保护对非并行意识的 FDW 免于陷入麻烦。
	 */
	fc_pstmt->subplans = NIL;
	foreach(fc_lc, fc_estate->es_plannedstmt->subplans)
	{
		Plan	   *fc_subplan = (Plan *) lfirst(fc_lc);

		if (fc_subplan && !fc_subplan->parallel_safe)
			fc_subplan = NULL;
		fc_pstmt->subplans = lappend(fc_pstmt->subplans, fc_subplan);
	}

	fc_pstmt->rewindPlanIDs = NULL;
	fc_pstmt->rowMarks = NIL;
	fc_pstmt->relationOids = NIL;
	fc_pstmt->invalItems = NIL;	/* 工作者无论如何无法重新规划…… */
	fc_pstmt->paramExecTypes = fc_estate->es_plannedstmt->paramExecTypes;
	fc_pstmt->utilityStmt = NULL;
	fc_pstmt->stmt_location = -1;
	fc_pstmt->stmt_len = -1;

	/* 返回我们的虚拟 PlannedStmt 的序列化副本。 */
	return nodeToString(fc_pstmt);
}

/*
 * 并行意识计划节点（有时还有其他节点）可能需要一些在所有并行工作者之间共享的状态。在我们调整 DSM 大小时，让它们有机会在 &pcxt->estimator 上调用 shm_toc_estimate_chunk 或 shm_toc_estimate_keys。
 *
 * 顺便说一句，计算树中 PlanState 节点的数量，以便我们知道需要多少 Instrumentation 结构。
 */
static bool fc_ExecParallelEstimate(PlanState *fc_planstate, ExecParallelEstimateContext *fc_e)
{
	if (fc_planstate == NULL)
		return false;

	/* 计算这个节点。 */
	fc_e->nnodes++;

	switch (nodeTag(fc_planstate))
	{
		case T_SeqScanState:
			if (fc_planstate->plan->parallel_aware)
				ExecSeqScanEstimate((SeqScanState *) fc_planstate,
									fc_e->pcxt);
			break;
		case T_IndexScanState:
			if (fc_planstate->plan->parallel_aware)
				ExecIndexScanEstimate((IndexScanState *) fc_planstate,
									  fc_e->pcxt);
			break;
		case T_IndexOnlyScanState:
			if (fc_planstate->plan->parallel_aware)
				ExecIndexOnlyScanEstimate((IndexOnlyScanState *) fc_planstate,
										  fc_e->pcxt);
			break;
		case T_ForeignScanState:
			if (fc_planstate->plan->parallel_aware)
				ExecForeignScanEstimate((ForeignScanState *) fc_planstate,
										fc_e->pcxt);
			break;
		case T_AppendState:
			if (fc_planstate->plan->parallel_aware)
				ExecAppendEstimate((AppendState *) fc_planstate,
								   fc_e->pcxt);
			break;
		case T_CustomScanState:
			if (fc_planstate->plan->parallel_aware)
				ExecCustomScanEstimate((CustomScanState *) fc_planstate,
									   fc_e->pcxt);
			break;
		case T_BitmapHeapScanState:
			if (fc_planstate->plan->parallel_aware)
				ExecBitmapHeapEstimate((BitmapHeapScanState *) fc_planstate,
									   fc_e->pcxt);
			break;
		case T_HashJoinState:
			if (fc_planstate->plan->parallel_aware)
				ExecHashJoinEstimate((HashJoinState *) fc_planstate,
									 fc_e->pcxt);
			break;
		case T_HashState:
			/* 即使在不并行意识的情况下，针对 EXPLAIN ANALYZE */
			ExecHashEstimate((HashState *) fc_planstate, fc_e->pcxt);
			break;
		case T_SortState:
			/* 即使在不并行意识的情况下，针对 EXPLAIN ANALYZE */
			ExecSortEstimate((SortState *) fc_planstate, fc_e->pcxt);
			break;
		case T_IncrementalSortState:
			/* 即使在不并行意识的情况下，针对 EXPLAIN ANALYZE */
			ExecIncrementalSortEstimate((IncrementalSortState *) fc_planstate, fc_e->pcxt);
			break;
		case T_AggState:
			/* 即使在不并行意识的情况下，针对 EXPLAIN ANALYZE */
			ExecAggEstimate((AggState *) fc_planstate, fc_e->pcxt);
			break;
		case T_MemoizeState:
			/* 即使在不并行意识的情况下，针对 EXPLAIN ANALYZE */
			ExecMemoizeEstimate((MemoizeState *) fc_planstate, fc_e->pcxt);
			break;
		default:
			break;
	}

	return planstate_tree_walker(fc_planstate, fc_ExecParallelEstimate, fc_e);
}

/*
 * 估计序列化指示参数所需的空间量。
 */
static Size fc_EstimateParamExecSpace(EState *fc_estate, Bitmapset *fc_params)
{
	int			fc_paramid;
	Size		fc_sz = sizeof(int);

	fc_paramid = -1;
	while ((fc_paramid = bms_next_member(fc_params, fc_paramid)) >= 0)
	{
		Oid			fc_typeOid;
		int16		fc_typLen;
		bool		fc_typByVal;
		ParamExecData *fc_prm;

		fc_prm = &(fc_estate->es_param_exec_vals[fc_paramid]);
		fc_typeOid = list_nth_oid(fc_estate->es_plannedstmt->paramExecTypes,
							   fc_paramid);

		fc_sz = add_size(fc_sz, sizeof(int)); /* 参数 id 的空间 */

		/* datum/isnull 的空间 */
		if (OidIsValid(fc_typeOid))
			get_typlenbyval(fc_typeOid, &fc_typLen, &fc_typByVal);
		else
		{
			/* 如果没有类型 OID，假设为按值，像 copyParamList 所做的那样。 */
			fc_typLen = sizeof(Datum);
			fc_typByVal = true;
		}
		fc_sz = add_size(fc_sz,
					  datumEstimateSpace(fc_prm->value, fc_prm->isnull,
										 fc_typByVal, fc_typLen));
	}
	return fc_sz;
}

/* 序列化指定的 PARAM_EXEC 参数。 */
/*
 * 我们首先写入参数的数量，作为一个 4 字节整数，然后
 * 依次写入每个参数的详细信息。每个参数的详细信息
 * 由一个 4 字节的 paramid（参数在执行时内部参数数组中的位置）
 * 和后续由 datumSerialize() 序列化的 datum 组成。
 */
static dsa_pointer
fc_SerializeParamExecParams(EState *fc_estate, Bitmapset *fc_params, dsa_area *fc_area)
{
	Size		fc_size;
	int			fc_nparams;
	int			fc_paramid;
	ParamExecData *fc_prm;
	dsa_pointer fc_handle;
	char	   *fc_start_address;

	/* 为当前参数值分配足够的空间。 */
	fc_size = fc_EstimateParamExecSpace(fc_estate, fc_params);
	fc_handle = dsa_allocate(fc_area, fc_size);
	fc_start_address = dsa_get_address(fc_area, fc_handle);

	/* 首先将参数数量写入作为 4 字节整数。 */
	fc_nparams = bms_num_members(fc_params);
	memcpy(fc_start_address, &fc_nparams, sizeof(int));
	fc_start_address += sizeof(int);

	/* 依次写入每个参数的详细信息。 */
	fc_paramid = -1;
	while ((fc_paramid = bms_next_member(fc_params, fc_paramid)) >= 0)
	{
		Oid			fc_typeOid;
		int16		fc_typLen;
		bool		fc_typByVal;

		fc_prm = &(fc_estate->es_param_exec_vals[fc_paramid]);
		fc_typeOid = list_nth_oid(fc_estate->es_plannedstmt->paramExecTypes,
							   fc_paramid);

		/* 写入 paramid。 */
		memcpy(fc_start_address, &fc_paramid, sizeof(int));
		fc_start_address += sizeof(int);

		/* 写入 datum/isnull */
		if (OidIsValid(fc_typeOid))
			get_typlenbyval(fc_typeOid, &fc_typLen, &fc_typByVal);
		else
		{
			/* 如果没有类型 OID，假设为按值，像 copyParamList 所做的那样。 */
			fc_typLen = sizeof(Datum);
			fc_typByVal = true;
		}
		datumSerialize(fc_prm->value, fc_prm->isnull, fc_typByVal, fc_typLen,
					   &fc_start_address);
	}

	return fc_handle;
}

/* 恢复指定的 PARAM_EXEC 参数。 */
static void fc_RestoreParamExecParams(char *fc_start_address, EState *fc_estate)
{
	int			fc_nparams;
	int			fc_i;
	int			fc_paramid;

	memcpy(&fc_nparams, fc_start_address, sizeof(int));
	fc_start_address += sizeof(int);

	for (fc_i = 0; fc_i < fc_nparams; fc_i++)
	{
		ParamExecData *fc_prm;

		/* 读取 paramid */
		memcpy(&fc_paramid, fc_start_address, sizeof(int));
		fc_start_address += sizeof(int);
		fc_prm = &(fc_estate->es_param_exec_vals[fc_paramid]);

		/* 读取 datum/isnull。 */
		fc_prm->value = datumRestore(&fc_start_address, &fc_prm->isnull);
		fc_prm->execPlan = NULL;
	}
}

/* 初始化将用于控制并行执行的动态共享内存段。 */
static bool fc_ExecParallelInitializeDSM(PlanState *fc_planstate,
						  ExecParallelInitializeDSMContext *fc_d)
{
	if (fc_planstate == NULL)
		return false;

	/* 如果启用了监控，则为此节点初始化插槽。 */
	if (fc_d->instrumentation != NULL)
		fc_d->instrumentation->plan_node_id[fc_d->nnodes] =
			fc_planstate->plan->plan_node_id;

	/* 计算这个节点。 */
	fc_d->nnodes++;

	/*
	 * 调用使用 DSM 的计划节点的初始化函数。
	 *
	 * 大多数计划节点在这里不会做任何事情，但分配了
	 * DSM 的计划节点可能需要在启动并行工作进程之前
	 * 初始化 DSM 中的共享状态。它们可以使用 shm_toc_allocate
	 * 分配之前估计的空间，并在每种情况下使用 shm_toc_insert
	 * 添加之前估计的键，目标是 pcxt->toc。
	 */
	switch (nodeTag(fc_planstate))
	{
		case T_SeqScanState:
			if (fc_planstate->plan->parallel_aware)
				ExecSeqScanInitializeDSM((SeqScanState *) fc_planstate,
										 fc_d->pcxt);
			break;
		case T_IndexScanState:
			if (fc_planstate->plan->parallel_aware)
				ExecIndexScanInitializeDSM((IndexScanState *) fc_planstate,
										   fc_d->pcxt);
			break;
		case T_IndexOnlyScanState:
			if (fc_planstate->plan->parallel_aware)
				ExecIndexOnlyScanInitializeDSM((IndexOnlyScanState *) fc_planstate,
											   fc_d->pcxt);
			break;
		case T_ForeignScanState:
			if (fc_planstate->plan->parallel_aware)
				ExecForeignScanInitializeDSM((ForeignScanState *) fc_planstate,
											 fc_d->pcxt);
			break;
		case T_AppendState:
			if (fc_planstate->plan->parallel_aware)
				ExecAppendInitializeDSM((AppendState *) fc_planstate,
										fc_d->pcxt);
			break;
		case T_CustomScanState:
			if (fc_planstate->plan->parallel_aware)
				ExecCustomScanInitializeDSM((CustomScanState *) fc_planstate,
											fc_d->pcxt);
			break;
		case T_BitmapHeapScanState:
			if (fc_planstate->plan->parallel_aware)
				ExecBitmapHeapInitializeDSM((BitmapHeapScanState *) fc_planstate,
											fc_d->pcxt);
			break;
		case T_HashJoinState:
			if (fc_planstate->plan->parallel_aware)
				ExecHashJoinInitializeDSM((HashJoinState *) fc_planstate,
										  fc_d->pcxt);
			break;
		case T_HashState:
			/* 即使在不并行意识的情况下，针对 EXPLAIN ANALYZE */
			ExecHashInitializeDSM((HashState *) fc_planstate, fc_d->pcxt);
			break;
		case T_SortState:
			/* 即使在不并行意识的情况下，针对 EXPLAIN ANALYZE */
			ExecSortInitializeDSM((SortState *) fc_planstate, fc_d->pcxt);
			break;
		case T_IncrementalSortState:
			/* 即使在不并行意识的情况下，针对 EXPLAIN ANALYZE */
			ExecIncrementalSortInitializeDSM((IncrementalSortState *) fc_planstate, fc_d->pcxt);
			break;
		case T_AggState:
			/* 即使在不并行意识的情况下，针对 EXPLAIN ANALYZE */
			ExecAggInitializeDSM((AggState *) fc_planstate, fc_d->pcxt);
			break;
		case T_MemoizeState:
			/* 即使在不并行意识的情况下，针对 EXPLAIN ANALYZE */
			ExecMemoizeInitializeDSM((MemoizeState *) fc_planstate, fc_d->pcxt);
			break;
		default:
			break;
	}

	return planstate_tree_walker(fc_planstate, fc_ExecParallelInitializeDSM, fc_d);
}

/*
 * 它为后端工作进程设置了响应队列，以便将元组
 * 返回给主后端并启动工作进程。
 */
static shm_mq_handle **
fc_ExecParallelSetupTupleQueues(ParallelContext *fc_pcxt, bool fc_reinitialize)
{
	shm_mq_handle **fc_responseq;
	char	   *fc_tqueuespace;
	int			fc_i;

	/* 如果没有工作进程，则跳过此步骤。 */
	if (fc_pcxt->nworkers == 0)
		return NULL;

	/* 为共享内存队列句柄分配内存。 */
	fc_responseq = (shm_mq_handle **)
		palloc(fc_pcxt->nworkers * sizeof(shm_mq_handle *));

	/*
	 * 如果不是重新初始化，则从 DSM 中为队列分配空间；
	 * 否则，查找已经分配的空间。
	 */
	if (!fc_reinitialize)
		fc_tqueuespace =
			shm_toc_allocate(fc_pcxt->toc,
							 mul_size(PARALLEL_TUPLE_QUEUE_SIZE,
									  fc_pcxt->nworkers));
	else
		fc_tqueuespace = shm_toc_lookup(fc_pcxt->toc, PARALLEL_KEY_TUPLE_QUEUE, false);

	/* 创建队列，并成为每个队列的接收者。 */
	for (fc_i = 0; fc_i < fc_pcxt->nworkers; ++fc_i)
	{
		shm_mq	   *fc_mq;

		fc_mq = shm_mq_create(fc_tqueuespace +
						   ((Size) fc_i) * PARALLEL_TUPLE_QUEUE_SIZE,
						   (Size) PARALLEL_TUPLE_QUEUE_SIZE);

		shm_mq_set_receiver(fc_mq, MyProc);
		fc_responseq[fc_i] = shm_mq_attach(fc_mq, fc_pcxt->seg, NULL);
	}

	/* 将队列数组添加到 shm_toc，以便其他人可以找到它。 */
	if (!fc_reinitialize)
		shm_toc_insert(fc_pcxt->toc, PARALLEL_KEY_TUPLE_QUEUE, fc_tqueuespace);

	/* 返回句柄数组。 */
	return fc_responseq;
}

/*
 * 为后端工作进程设置执行的必要基础设施
 * 并将结果返回给主后端。
 */
ParallelExecutorInfo *
ExecInitParallelPlan(PlanState *fc_planstate, EState *fc_estate,
					 Bitmapset *fc_sendParams, int fc_nworkers,
					 int64 fc_tuples_needed)
{
	ParallelExecutorInfo *fc_pei;
	ParallelContext *fc_pcxt;
	ExecParallelEstimateContext fc_e;
	ExecParallelInitializeDSMContext fc_d;
	FixedParallelExecutorState *fc_fpes;
	char	   *fc_pstmt_data;
	char	   *fc_pstmt_space;
	char	   *fc_paramlistinfo_space;
	BufferUsage *fc_bufusage_space;
	WalUsage   *fc_walusage_space;
	SharedExecutorInstrumentation *fc_instrumentation = NULL;
	SharedJitInstrumentation *fc_jit_instrumentation = NULL;
	int			fc_pstmt_len;
	int			fc_paramlistinfo_len;
	int			fc_instrumentation_len = 0;
	int			fc_jit_instrumentation_len = 0;
	int			fc_instrument_offset = 0;
	Size		fc_dsa_minsize = dsa_minimum_size();
	char	   *fc_query_string;
	int			fc_query_len;

	/*
	 * 强制评估将传递给工作进程的任何 initplan 输出，
	 * 如果它们尚未被评估。
	 *
	 * 为简单起见，我们在这里使用 EState 的每个输出元组
	 * 的 ExprContext。这存在查询内部内存泄漏的风险，
	 * 因为在 ExprContext 被重置之前，我们可能会多次经过这里；
	 * 但是 ExecSetParamPlan 通常不会在上下文中泄漏任何内存
	 * （参见其注释），因此这似乎不值得复杂化此函数的 API
	 * 以传递一个生命周期更短的 ExprContext。将来可能需要改变。
	 */
	ExecSetParamPlanMulti(fc_sendParams, GetPerTupleExprContext(fc_estate));

	/* 为返回值分配对象。 */
	fc_pei = palloc0(sizeof(ParallelExecutorInfo));
	fc_pei->finished = false;
	fc_pei->planstate = fc_planstate;

	/* 修复并序列化要发送给工作进程的计划。 */
	fc_pstmt_data = fc_ExecSerializePlan(fc_planstate->plan, fc_estate);

	/* 创建并行上下文。 */
	fc_pcxt = CreateParallelContext("postgres", "ParallelQueryMain", fc_nworkers);
	fc_pei->pcxt = fc_pcxt;

	/*
	 * 在告诉并行上下文创建一个动态共享内存段之前，
	 * 我们需要弄清楚它应该有多大。估计我们需要存储的
	 * 各种东西的空间。
	 */

	/* 估计固定大小状态的空间。 */
	shm_toc_estimate_chunk(&fc_pcxt->estimator,
						   sizeof(FixedParallelExecutorState));
	shm_toc_estimate_keys(&fc_pcxt->estimator, 1);

	/* 估计查询文本的空间。 */
	fc_query_len = strlen(fc_estate->es_sourceText);
	shm_toc_estimate_chunk(&fc_pcxt->estimator, fc_query_len + 1);
	shm_toc_estimate_keys(&fc_pcxt->estimator, 1);

	/* 估计序列化 PlannedStmt 的空间。 */
	fc_pstmt_len = strlen(fc_pstmt_data) + 1;
	shm_toc_estimate_chunk(&fc_pcxt->estimator, fc_pstmt_len);
	shm_toc_estimate_keys(&fc_pcxt->estimator, 1);

	/* 估计序列化 ParamListInfo 的空间。 */
	fc_paramlistinfo_len = EstimateParamListSpace(fc_estate->es_param_list_info);
	shm_toc_estimate_chunk(&fc_pcxt->estimator, fc_paramlistinfo_len);
	shm_toc_estimate_keys(&fc_pcxt->estimator, 1);

	/*
	 * 估计 BufferUsage 的空间。
	 *
	 * 如果没有使用 EXPLAIN，并且没有加载任何相关的扩展，
	 * 那么我们可以跳过这个。但我们无法知道是否有人在
	 * 查看 pgBufferUsage，因此无条件执行这一操作。
	 */
	shm_toc_estimate_chunk(&fc_pcxt->estimator,
						   mul_size(sizeof(BufferUsage), fc_pcxt->nworkers));
	shm_toc_estimate_keys(&fc_pcxt->estimator, 1);

	/*
	 * WalUsage 也是同样的情况。
	 */
	shm_toc_estimate_chunk(&fc_pcxt->estimator,
						   mul_size(sizeof(WalUsage), fc_pcxt->nworkers));
	shm_toc_estimate_keys(&fc_pcxt->estimator, 1);

	/* 估计元组队列的空间。 */
	shm_toc_estimate_chunk(&fc_pcxt->estimator,
						   mul_size(PARALLEL_TUPLE_QUEUE_SIZE, fc_pcxt->nworkers));
	shm_toc_estimate_keys(&fc_pcxt->estimator, 1);

	/*
	 * 给并行感知节点一个机会来增加估计，并获得有多少个 PlanState 节点的计数。
	 */
	fc_e.pcxt = fc_pcxt;
	fc_e.nnodes = 0;
	fc_ExecParallelEstimate(fc_planstate, &fc_e);

	/* 估计所需的仪器空间。 */
	if (fc_estate->es_instrument)
	{
		fc_instrumentation_len =
			offsetof(SharedExecutorInstrumentation, plan_node_id) +
			sizeof(int) * fc_e.nnodes;
		fc_instrumentation_len = MAXALIGN(fc_instrumentation_len);
		fc_instrument_offset = fc_instrumentation_len;
		fc_instrumentation_len +=
			mul_size(sizeof(Instrumentation),
					 mul_size(fc_e.nnodes, fc_nworkers));
		shm_toc_estimate_chunk(&fc_pcxt->estimator, fc_instrumentation_len);
		shm_toc_estimate_keys(&fc_pcxt->estimator, 1);

		/* 估计所需的 JIT 仪器空间。 */
		if (fc_estate->es_jit_flags != PGJIT_NONE)
		{
			fc_jit_instrumentation_len =
				offsetof(SharedJitInstrumentation, jit_instr) +
				sizeof(JitInstrumentation) * fc_nworkers;
			shm_toc_estimate_chunk(&fc_pcxt->estimator, fc_jit_instrumentation_len);
			shm_toc_estimate_keys(&fc_pcxt->estimator, 1);
		}
	}

	/* 估计 DSA 区域的空间。 */
	shm_toc_estimate_chunk(&fc_pcxt->estimator, fc_dsa_minsize);
	shm_toc_estimate_keys(&fc_pcxt->estimator, 1);

	/* 每个人都有机会请求空间，所以现在创建 DSM。 */
	InitializeParallelDSM(fc_pcxt);

	/*
	 * 好的，现在我们有一个动态共享内存段，它应该足够大
	 * 以存储我们估计想要放入的数据，以及 ParallelContext 自身
	 * 需要存储的其他一般内容（并非专门与执行器相关）。不过，
	 * 我们所申请的空间尚未分配或初始化，所以得进行这一步。
	 */

	/* 存储固定大小的状态。 */
	fc_fpes = shm_toc_allocate(fc_pcxt->toc, sizeof(FixedParallelExecutorState));
	fc_fpes->tuples_needed = fc_tuples_needed;
	fc_fpes->param_exec = InvalidDsaPointer;
	fc_fpes->eflags = fc_estate->es_top_eflags;
	fc_fpes->jit_flags = fc_estate->es_jit_flags;
	shm_toc_insert(fc_pcxt->toc, PARALLEL_KEY_EXECUTOR_FIXED, fc_fpes);

	/* 存储查询字符串 */
	fc_query_string = shm_toc_allocate(fc_pcxt->toc, fc_query_len + 1);
	memcpy(fc_query_string, fc_estate->es_sourceText, fc_query_len + 1);
	shm_toc_insert(fc_pcxt->toc, PARALLEL_KEY_QUERY_TEXT, fc_query_string);

	/* 存储序列化的 PlannedStmt。 */
	fc_pstmt_space = shm_toc_allocate(fc_pcxt->toc, fc_pstmt_len);
	memcpy(fc_pstmt_space, fc_pstmt_data, fc_pstmt_len);
	shm_toc_insert(fc_pcxt->toc, PARALLEL_KEY_PLANNEDSTMT, fc_pstmt_space);

	/* 存储序列化的 ParamListInfo。 */
	fc_paramlistinfo_space = shm_toc_allocate(fc_pcxt->toc, fc_paramlistinfo_len);
	shm_toc_insert(fc_pcxt->toc, PARALLEL_KEY_PARAMLISTINFO, fc_paramlistinfo_space);
	SerializeParamList(fc_estate->es_param_list_info, &fc_paramlistinfo_space);

	/* 为每个工作者的 BufferUsage 分配空间；不需要初始化。 */
	fc_bufusage_space = shm_toc_allocate(fc_pcxt->toc,
									  mul_size(sizeof(BufferUsage), fc_pcxt->nworkers));
	shm_toc_insert(fc_pcxt->toc, PARALLEL_KEY_BUFFER_USAGE, fc_bufusage_space);
	fc_pei->buffer_usage = fc_bufusage_space;

	/* 对于 WalUsage 也是如此。 */
	fc_walusage_space = shm_toc_allocate(fc_pcxt->toc,
									  mul_size(sizeof(WalUsage), fc_pcxt->nworkers));
	shm_toc_insert(fc_pcxt->toc, PARALLEL_KEY_WAL_USAGE, fc_walusage_space);
	fc_pei->wal_usage = fc_walusage_space;

	/* 设置工作者将要写入的元组队列。 */
	fc_pei->tqueue = fc_ExecParallelSetupTupleQueues(fc_pcxt, false);

	/* 不过，我们暂时不需要 TupleQueueReaders。 */
	fc_pei->reader = NULL;

	/*
	 * 如果提供了仪器选项，则为数据分配空间。
	 * 这里只部分初始化其内容；其余的将在 ExecParallelInitializeDSM 中完成。
	 */
	if (fc_estate->es_instrument)
	{
		Instrumentation *fc_instrument;
		int			fc_i;

		fc_instrumentation = shm_toc_allocate(fc_pcxt->toc, fc_instrumentation_len);
		fc_instrumentation->instrument_options = fc_estate->es_instrument;
		fc_instrumentation->instrument_offset = fc_instrument_offset;
		fc_instrumentation->num_workers = fc_nworkers;
		fc_instrumentation->num_plan_nodes = fc_e.nnodes;
		fc_instrument = GetInstrumentationArray(fc_instrumentation);
		for (fc_i = 0; fc_i < fc_nworkers * fc_e.nnodes; ++fc_i)
			InstrInit(&fc_instrument[fc_i], fc_estate->es_instrument);
		shm_toc_insert(fc_pcxt->toc, PARALLEL_KEY_INSTRUMENTATION,
					   fc_instrumentation);
		fc_pei->instrumentation = fc_instrumentation;

		if (fc_estate->es_jit_flags != PGJIT_NONE)
		{
			fc_jit_instrumentation = shm_toc_allocate(fc_pcxt->toc,
												   fc_jit_instrumentation_len);
			fc_jit_instrumentation->num_workers = fc_nworkers;
			memset(fc_jit_instrumentation->jit_instr, 0,
				   sizeof(JitInstrumentation) * fc_nworkers);
			shm_toc_insert(fc_pcxt->toc, PARALLEL_KEY_JIT_INSTRUMENTATION,
						   fc_jit_instrumentation);
			fc_pei->jit_instrumentation = fc_jit_instrumentation;
		}
	}

	/*
	 * 创建一个可以由领导者和所有工作者使用的 DSA 区域。
	 * （然而，如果我们未能创建 DSM 而是使用私有内存，那么跳过这一步。）
	 */
	if (fc_pcxt->seg != NULL)
	{
		char	   *fc_area_space;

		fc_area_space = shm_toc_allocate(fc_pcxt->toc, fc_dsa_minsize);
		shm_toc_insert(fc_pcxt->toc, PARALLEL_KEY_DSA, fc_area_space);
		fc_pei->area = dsa_create_in_place(fc_area_space, fc_dsa_minsize,
										LWTRANCHE_PARALLEL_QUERY_DSA,
										fc_pcxt->seg);

		/*
		 * 使用 DSA 存储序列化参数（如果有）。
		 * 因为我们可能在值改变后重新启动工作者（因此所需的
		 * 存储量也发生变化），所以我们不敢使用主并行查询的 DSM。
		 */
		if (!bms_is_empty(fc_sendParams))
		{
			fc_pei->param_exec = fc_SerializeParamExecParams(fc_estate, fc_sendParams,
													   fc_pei->area);
			fc_fpes->param_exec = fc_pei->param_exec;
		}
	}

	/*
	 * 给并行感知节点一个机会来初始化它们的共享数据。
	 * 如果存在，这也会初始化 instrumentation->ps_instrument 的元素。
	 */
	fc_d.pcxt = fc_pcxt;
	fc_d.instrumentation = fc_instrumentation;
	fc_d.nnodes = 0;

	/* 在初始化计划时安装我们的 DSA 区域。 */
	fc_estate->es_query_dsa = fc_pei->area;
	fc_ExecParallelInitializeDSM(fc_planstate, &fc_d);
	fc_estate->es_query_dsa = NULL;

	/*
	 * 确保世界没有在我们脚下发生变化。 这可能
	 * 仅仅是一个 Assert()，不过我们现在还是保守点。
	 */
	if (fc_e.nnodes != fc_d.nnodes)
		elog(ERROR, "inconsistent count of PlanState nodes");

	/* 好吧，我们准备开始了。 */
	return fc_pei;
}

/*
 * 设置元组队列读取器，以读取并行子计划的结果。
 *
 * 这与 ExecInitParallelPlan() 是分开的，因为我们可以启动工作
 * 进程并让它们开始执行某些操作，然后再执行这一步。
 */
void ExecParallelCreateReaders(ParallelExecutorInfo *fc_pei)
{
	int			fc_nworkers = fc_pei->pcxt->nworkers_launched;
	int			fc_i;

	Assert(fc_pei->reader == NULL);

	if (fc_nworkers > 0)
	{
		fc_pei->reader = (TupleQueueReader **)
			palloc(fc_nworkers * sizeof(TupleQueueReader *));

		for (fc_i = 0; fc_i < fc_nworkers; fc_i++)
		{
			shm_mq_set_handle(fc_pei->tqueue[fc_i],
							  fc_pei->pcxt->worker[fc_i].bgwhandle);
			fc_pei->reader[fc_i] = CreateTupleQueueReader(fc_pei->tqueue[fc_i]);
		}
	}
}

/*
 * 在启动新一批工作者之前重新初始化并行执行器的共享内存状态。
 */
void ExecParallelReinitialize(PlanState *fc_planstate,
						 ParallelExecutorInfo *fc_pei,
						 Bitmapset *fc_sendParams)
{
	EState	   *fc_estate = fc_planstate->state;
	FixedParallelExecutorState *fc_fpes;

	/* 旧工作者必须已经关闭 */
	Assert(fc_pei->finished);

	/*
	 * 强迫任何将要传递给工作者的 initplan 输出被
	 * 评估，如果它们尚未被评估（参见 ExecInitParallelPlan 中的注释）。
	 */
	ExecSetParamPlanMulti(fc_sendParams, GetPerTupleExprContext(fc_estate));

	ReinitializeParallelDSM(fc_pei->pcxt);
	fc_pei->tqueue = fc_ExecParallelSetupTupleQueues(fc_pei->pcxt, true);
	fc_pei->reader = NULL;
	fc_pei->finished = false;

	fc_fpes = shm_toc_lookup(fc_pei->pcxt->toc, PARALLEL_KEY_EXECUTOR_FIXED, false);

	/* 释放上一轮的任何序列化参数。 */
	if (DsaPointerIsValid(fc_fpes->param_exec))
	{
		dsa_free(fc_pei->area, fc_fpes->param_exec);
		fc_fpes->param_exec = InvalidDsaPointer;
	}

	/* 如果需要，序列化当前参数值。 */
	if (!bms_is_empty(fc_sendParams))
	{
		fc_pei->param_exec = fc_SerializeParamExecParams(fc_estate, fc_sendParams,
												   fc_pei->area);
		fc_fpes->param_exec = fc_pei->param_exec;
	}

	/* traversal plan tree and let each child node reset associated state. */
	fc_estate->es_query_dsa = fc_pei->area;
	fc_ExecParallelReInitializeDSM(fc_planstate, fc_pei->pcxt);
	fc_estate->es_query_dsa = NULL;
}

/*
 * 遍历计划树以重新初始化每个节点的动态共享内存状态
 */
static bool fc_ExecParallelReInitializeDSM(PlanState *fc_planstate,
							ParallelContext *fc_pcxt)
{
	if (fc_planstate == NULL)
		return false;

	/*
	 * 调用使用 DSM 的计划节点的重新初始化函数。
	 */
	switch (nodeTag(fc_planstate))
	{
		case T_SeqScanState:
			if (fc_planstate->plan->parallel_aware)
				ExecSeqScanReInitializeDSM((SeqScanState *) fc_planstate,
										   fc_pcxt);
			break;
		case T_IndexScanState:
			if (fc_planstate->plan->parallel_aware)
				ExecIndexScanReInitializeDSM((IndexScanState *) fc_planstate,
											 fc_pcxt);
			break;
		case T_IndexOnlyScanState:
			if (fc_planstate->plan->parallel_aware)
				ExecIndexOnlyScanReInitializeDSM((IndexOnlyScanState *) fc_planstate,
												 fc_pcxt);
			break;
		case T_ForeignScanState:
			if (fc_planstate->plan->parallel_aware)
				ExecForeignScanReInitializeDSM((ForeignScanState *) fc_planstate,
											   fc_pcxt);
			break;
		case T_AppendState:
			if (fc_planstate->plan->parallel_aware)
				ExecAppendReInitializeDSM((AppendState *) fc_planstate, fc_pcxt);
			break;
		case T_CustomScanState:
			if (fc_planstate->plan->parallel_aware)
				ExecCustomScanReInitializeDSM((CustomScanState *) fc_planstate,
											  fc_pcxt);
			break;
		case T_BitmapHeapScanState:
			if (fc_planstate->plan->parallel_aware)
				ExecBitmapHeapReInitializeDSM((BitmapHeapScanState *) fc_planstate,
											  fc_pcxt);
			break;
		case T_HashJoinState:
			if (fc_planstate->plan->parallel_aware)
				ExecHashJoinReInitializeDSM((HashJoinState *) fc_planstate,
											fc_pcxt);
			break;
		case T_HashState:
		case T_SortState:
		case T_IncrementalSortState:
		case T_MemoizeState:
			/* 这些节点具有 DSM 状态，但不需要重新初始化 */
			break;

		default:
			break;
	}

	return planstate_tree_walker(fc_planstate, fc_ExecParallelReInitializeDSM, fc_pcxt);
}

/*
 * 从动态共享内存中复制有关该节点及其后代的仪器信息。
 */
static bool fc_ExecParallelRetrieveInstrumentation(PlanState *fc_planstate,
									SharedExecutorInstrumentation *fc_instrumentation)
{
	Instrumentation *fc_instrument;
	int			fc_i;
	int			fc_n;
	int			fc_ibytes;
	int			fc_plan_node_id = fc_planstate->plan->plan_node_id;
	MemoryContext fc_oldcontext;

	/* 查找该节点的仪器信息。 */
	for (fc_i = 0; fc_i < fc_instrumentation->num_plan_nodes; ++fc_i)
		if (fc_instrumentation->plan_node_id[fc_i] == fc_plan_node_id)
			break;
	if (fc_i >= fc_instrumentation->num_plan_nodes)
		elog(ERROR, "plan node %d not found", fc_plan_node_id);

	/* 汇总所有工作线程的统计信息。 */
	fc_instrument = GetInstrumentationArray(fc_instrumentation);
	fc_instrument += fc_i * fc_instrumentation->num_workers;
	for (fc_n = 0; fc_n < fc_instrumentation->num_workers; ++fc_n)
		InstrAggNode(fc_planstate->instrument, &fc_instrument[fc_n]);

	/*
	 * 还存储每个工作线程的详细信息。
	 *
	 * 工作线程的仪器信息应在与常规仪器信息相同的上下文中分配，即每个查询上下文。
	 * 切换到每个查询内存上下文。
	 */
	fc_oldcontext = MemoryContextSwitchTo(fc_planstate->state->es_query_cxt);
	fc_ibytes = mul_size(fc_instrumentation->num_workers, sizeof(Instrumentation));
	fc_planstate->worker_instrument =
		palloc(fc_ibytes + offsetof(WorkerInstrumentation, instrument));
	MemoryContextSwitchTo(fc_oldcontext);

	fc_planstate->worker_instrument->num_workers = fc_instrumentation->num_workers;
	memcpy(&fc_planstate->worker_instrument->instrument, fc_instrument, fc_ibytes);

	/* 执行任何特定于节点类型的工作。 */
	switch (nodeTag(fc_planstate))
	{
		case T_SortState:
			ExecSortRetrieveInstrumentation((SortState *) fc_planstate);
			break;
		case T_IncrementalSortState:
			ExecIncrementalSortRetrieveInstrumentation((IncrementalSortState *) fc_planstate);
			break;
		case T_HashState:
			ExecHashRetrieveInstrumentation((HashState *) fc_planstate);
			break;
		case T_AggState:
			ExecAggRetrieveInstrumentation((AggState *) fc_planstate);
			break;
		case T_MemoizeState:
			ExecMemoizeRetrieveInstrumentation((MemoizeState *) fc_planstate);
			break;
		default:
			break;
	}

	return planstate_tree_walker(fc_planstate, fc_ExecParallelRetrieveInstrumentation,
								 fc_instrumentation);
}

/*
 * 从动态共享内存中累加工作线程的 JIT 仪器信息。
 */
static void fc_ExecParallelRetrieveJitInstrumentation(PlanState *fc_planstate,
									   SharedJitInstrumentation *fc_shared_jit)
{
	JitInstrumentation *fc_combined;
	int			fc_ibytes;

	int			fc_n;

	/*
	 * 将工作线程的 JIT 仪器信息汇总到组合的 JIT
	 * 仪器信息中，如果需要则分配它。
	 */
	if (!fc_planstate->state->es_jit_worker_instr)
		fc_planstate->state->es_jit_worker_instr =
			MemoryContextAllocZero(fc_planstate->state->es_query_cxt, sizeof(JitInstrumentation));
	fc_combined = fc_planstate->state->es_jit_worker_instr;

	/* 汇总所有工作线程的仪器信息。 */
	for (fc_n = 0; fc_n < fc_shared_jit->num_workers; ++fc_n)
		InstrJitAgg(fc_combined, &fc_shared_jit->jit_instr[fc_n]);

	/*
	 * 存储每个工作线程的详细信息。
	 *
	 * 类似于 ExecParallelRetrieveInstrumentation()，在每个查询上下文中分配仪器信息。
	 */
	fc_ibytes = offsetof(SharedJitInstrumentation, jit_instr)
		+ mul_size(fc_shared_jit->num_workers, sizeof(JitInstrumentation));
	fc_planstate->worker_jit_instrument =
		MemoryContextAlloc(fc_planstate->state->es_query_cxt, fc_ibytes);

	memcpy(fc_planstate->worker_jit_instrument, fc_shared_jit, fc_ibytes);
}

/*
 * 完成并行执行。我们等待并行工作线程完成，并
 * 汇总它们的缓冲区/WAL 使用情况。
 */
void ExecParallelFinish(ParallelExecutorInfo *fc_pei)
{
	int			fc_nworkers = fc_pei->pcxt->nworkers_launched;
	int			fc_i;

	/* 如果连续调用两次，则将此操作视为无操作。 */
	if (fc_pei->finished)
		return;

	/*
	 * 尽快从元组队列中分离，以便任何仍处于活动状态的工作线程将
	 * 注意到不再需要进一步的结果。
	 */
	if (fc_pei->tqueue != NULL)
	{
		for (fc_i = 0; fc_i < fc_nworkers; fc_i++)
			shm_mq_detach(fc_pei->tqueue[fc_i]);
		pfree(fc_pei->tqueue);
		fc_pei->tqueue = NULL;
	}

	/*
	 * 在等待工作线程完成时，让我们摆脱
	 * 元组队列读取器。 （此处也可以进行其他本地清理。）
	 */
	if (fc_pei->reader != NULL)
	{
		for (fc_i = 0; fc_i < fc_nworkers; fc_i++)
			DestroyTupleQueueReader(fc_pei->reader[fc_i]);
		pfree(fc_pei->reader);
		fc_pei->reader = NULL;
	}

	/* 现在等待工作线程完成。 */
	WaitForParallelWorkersToFinish(fc_pei->pcxt);

	/*
	 * 接下来，汇总缓冲区/WAL 使用情况。 （这必须等到工作线程完成，
	 * 否则我们可能会得到不完整的数据。）
	 */
	for (fc_i = 0; fc_i < fc_nworkers; fc_i++)
		InstrAccumParallelQuery(&fc_pei->buffer_usage[fc_i], &fc_pei->wal_usage[fc_i]);

	fc_pei->finished = true;
}

/*
 * 汇总仪器信息，然后清理 ExecParallelFinish 后仍然存在的任何 ParallelExecutorInfo
 * 资源。我们将这些例程分开，因为可能有人希望检查在 ExecParallelFinish
 * 之后和调用该例程之前 DSM 的内容。
 */
void ExecParallelCleanup(ParallelExecutorInfo *fc_pei)
{
	/* 汇总仪器信息（如果有）。 */
	if (fc_pei->instrumentation)
		fc_ExecParallelRetrieveInstrumentation(fc_pei->planstate,
											fc_pei->instrumentation);

	/* 汇总 JIT 仪器信息（如果有）。 */
	if (fc_pei->jit_instrumentation)
		fc_ExecParallelRetrieveJitInstrumentation(fc_pei->planstate,
											   fc_pei->jit_instrumentation);

	/* 释放任何序列化参数。 */
	if (DsaPointerIsValid(fc_pei->param_exec))
	{
		dsa_free(fc_pei->area, fc_pei->param_exec);
		fc_pei->param_exec = InvalidDsaPointer;
	}
	if (fc_pei->area != NULL)
	{
		dsa_detach(fc_pei->area);
		fc_pei->area = NULL;
	}
	if (fc_pei->pcxt != NULL)
	{
		DestroyParallelContext(fc_pei->pcxt);
		fc_pei->pcxt = NULL;
	}
	pfree(fc_pei);
}

/*
 * 创建一个 DestReceiver，将我们生成的元组写入为此目的而指定的 shm_mq。
 */
static DestReceiver *
fc_ExecParallelGetReceiver(dsm_segment *fc_seg, shm_toc *fc_toc)
{
	char	   *fc_mqspace;
	shm_mq	   *fc_mq;

	fc_mqspace = shm_toc_lookup(fc_toc, PARALLEL_KEY_TUPLE_QUEUE, false);
	fc_mqspace += ParallelWorkerNumber * PARALLEL_TUPLE_QUEUE_SIZE;
	fc_mq = (shm_mq *) fc_mqspace;
	shm_mq_set_sender(fc_mq, MyProc);
	return CreateTupleQueueDestReceiver(shm_mq_attach(fc_mq, fc_seg, NULL));
}

/*
 * 为我们要执行的 PlannedStmt 创建一个 QueryDesc，并返回它。
 */
static QueryDesc *
fc_ExecParallelGetQueryDesc(shm_toc *fc_toc, DestReceiver *fc_receiver,
						 int fc_instrument_options)
{
	char	   *fc_pstmtspace;
	char	   *fc_paramspace;
	PlannedStmt *fc_pstmt;
	ParamListInfo fc_paramLI;
	char	   *fc_queryString;

	/* 从共享内存中获取查询字符串 */
	fc_queryString = shm_toc_lookup(fc_toc, PARALLEL_KEY_QUERY_TEXT, false);

	/* 重构领导者提供的 PlannedStmt。 */
	fc_pstmtspace = shm_toc_lookup(fc_toc, PARALLEL_KEY_PLANNEDSTMT, false);
	fc_pstmt = (PlannedStmt *) stringToNode(fc_pstmtspace);

	/* 重构 ParamListInfo。 */
	fc_paramspace = shm_toc_lookup(fc_toc, PARALLEL_KEY_PARAMLISTINFO, false);
	fc_paramLI = RestoreParamList(&fc_paramspace);

	/* 为查询创建一个 QueryDesc。 */
	return CreateQueryDesc(fc_pstmt,
						   fc_queryString,
						   GetActiveSnapshot(), InvalidSnapshot,
						   fc_receiver, fc_paramLI, NULL, fc_instrument_options);
}

/*
 * 从该节点及其后代复制仪器信息到动态共享内存，以便并行领导者可以检索到它。
 */
static bool fc_ExecParallelReportInstrumentation(PlanState *fc_planstate,
								  SharedExecutorInstrumentation *fc_instrumentation)
{
	int			fc_i;
	int			fc_plan_node_id = fc_planstate->plan->plan_node_id;
	Instrumentation *fc_instrument;

	InstrEndLoop(fc_planstate->instrument);

	/*
	 * 如果我们将 ps_instrument 中的 plan_node_id 值洗牌为排序
	 * 顺序，我们可以在这里使用二分搜索。如果我们推送足够大的计划树
	 * ，这可能会变得重要。现在，还是采用慢而笨的方法。
	 */
	for (fc_i = 0; fc_i < fc_instrumentation->num_plan_nodes; ++fc_i)
		if (fc_instrumentation->plan_node_id[fc_i] == fc_plan_node_id)
			break;
	if (fc_i >= fc_instrumentation->num_plan_nodes)
		elog(ERROR, "plan node %d not found", fc_plan_node_id);

	/*
	 * 将我们的统计数据添加到每个节点、每个工作进程的总计中。
	 * 如果我们重新启动工作进程，这可能会发生多次。
	 */
	fc_instrument = GetInstrumentationArray(fc_instrumentation);
	fc_instrument += fc_i * fc_instrumentation->num_workers;
	Assert(IsParallelWorker());
	Assert(ParallelWorkerNumber < fc_instrumentation->num_workers);
	InstrAggNode(&fc_instrument[ParallelWorkerNumber], fc_planstate->instrument);

	return planstate_tree_walker(fc_planstate, fc_ExecParallelReportInstrumentation,
								 fc_instrumentation);
}

/*
 * 使用从共享内存中检索到的信息初始化 PlanState 及其后代。
 * 这必须在 Executor 分配和初始化 PlanState 之后完成；
 * 也就是说，在 ExecutorStart() 之后。
 */
static bool fc_ExecParallelInitializeWorker(PlanState *fc_planstate, ParallelWorkerContext *fc_pwcxt)
{
	if (fc_planstate == NULL)
		return false;

	switch (nodeTag(fc_planstate))
	{
		case T_SeqScanState:
			if (fc_planstate->plan->parallel_aware)
				ExecSeqScanInitializeWorker((SeqScanState *) fc_planstate, fc_pwcxt);
			break;
		case T_IndexScanState:
			if (fc_planstate->plan->parallel_aware)
				ExecIndexScanInitializeWorker((IndexScanState *) fc_planstate,
											  fc_pwcxt);
			break;
		case T_IndexOnlyScanState:
			if (fc_planstate->plan->parallel_aware)
				ExecIndexOnlyScanInitializeWorker((IndexOnlyScanState *) fc_planstate,
												  fc_pwcxt);
			break;
		case T_ForeignScanState:
			if (fc_planstate->plan->parallel_aware)
				ExecForeignScanInitializeWorker((ForeignScanState *) fc_planstate,
												fc_pwcxt);
			break;
		case T_AppendState:
			if (fc_planstate->plan->parallel_aware)
				ExecAppendInitializeWorker((AppendState *) fc_planstate, fc_pwcxt);
			break;
		case T_CustomScanState:
			if (fc_planstate->plan->parallel_aware)
				ExecCustomScanInitializeWorker((CustomScanState *) fc_planstate,
											   fc_pwcxt);
			break;
		case T_BitmapHeapScanState:
			if (fc_planstate->plan->parallel_aware)
				ExecBitmapHeapInitializeWorker((BitmapHeapScanState *) fc_planstate,
											   fc_pwcxt);
			break;
		case T_HashJoinState:
			if (fc_planstate->plan->parallel_aware)
				ExecHashJoinInitializeWorker((HashJoinState *) fc_planstate,
											 fc_pwcxt);
			break;
		case T_HashState:
			/* 即使在不并行意识的情况下，针对 EXPLAIN ANALYZE */
			ExecHashInitializeWorker((HashState *) fc_planstate, fc_pwcxt);
			break;
		case T_SortState:
			/* 即使在不并行意识的情况下，针对 EXPLAIN ANALYZE */
			ExecSortInitializeWorker((SortState *) fc_planstate, fc_pwcxt);
			break;
		case T_IncrementalSortState:
			/* 即使在不并行意识的情况下，针对 EXPLAIN ANALYZE */
			ExecIncrementalSortInitializeWorker((IncrementalSortState *) fc_planstate,
												fc_pwcxt);
			break;
		case T_AggState:
			/* 即使在不并行意识的情况下，针对 EXPLAIN ANALYZE */
			ExecAggInitializeWorker((AggState *) fc_planstate, fc_pwcxt);
			break;
		case T_MemoizeState:
			/* 即使在不并行意识的情况下，针对 EXPLAIN ANALYZE */
			ExecMemoizeInitializeWorker((MemoizeState *) fc_planstate, fc_pwcxt);
			break;
		default:
			break;
	}

	return planstate_tree_walker(fc_planstate, fc_ExecParallelInitializeWorker,
								 fc_pwcxt);
}

/*
 * 并行查询工作进程的主要入口点。
 *
 * 我们从 ParallelWorkerMain 到达这个函数，因此创建一个合理
 * 并行环境所需的设置已经完成；ParallelWorkerMain 处理事务
 * 状态、组合 CID 映射和 GUC 值等问题，所以我们不需要在
 * 这里处理这些。
 *
 * 我们的工作是处理与执行器相关的特定问题。并行组领导者会
 * 存储一个序列化的 PlannedStmt，而我们的工作是执行该计划
 * 并将结果元组写入适当的元组队列。我们还需要的一些支持信息
 * 也存储在 dsm_segment 中，可以通过 shm_toc 访问。
 */
void ParallelQueryMain(dsm_segment *fc_seg, shm_toc *fc_toc)
{
	FixedParallelExecutorState *fc_fpes;
	BufferUsage *fc_buffer_usage;
	WalUsage   *fc_wal_usage;
	DestReceiver *fc_receiver;
	QueryDesc  *fc_queryDesc;
	SharedExecutorInstrumentation *fc_instrumentation;
	SharedJitInstrumentation *fc_jit_instrumentation;
	int			fc_instrument_options = 0;
	void	   *fc_area_space;
	dsa_area   *fc_area;
	ParallelWorkerContext fc_pwcxt;

	/* 获取固定大小的状态。 */
	fc_fpes = shm_toc_lookup(fc_toc, PARALLEL_KEY_EXECUTOR_FIXED, false);

	/* 设置 DestReceiver、SharedExecutorInstrumentation 和 QueryDesc。 */
	fc_receiver = fc_ExecParallelGetReceiver(fc_seg, fc_toc);
	fc_instrumentation = shm_toc_lookup(fc_toc, PARALLEL_KEY_INSTRUMENTATION, true);
	if (fc_instrumentation != NULL)
		fc_instrument_options = fc_instrumentation->instrument_options;
	fc_jit_instrumentation = shm_toc_lookup(fc_toc, PARALLEL_KEY_JIT_INSTRUMENTATION,
										 true);
	fc_queryDesc = fc_ExecParallelGetQueryDesc(fc_toc, fc_receiver, fc_instrument_options);

	/* 设置个别工作进程的 debug_query_string */
	debug_query_string = fc_queryDesc->sourceText;

	/* 报告工作进程的查询以供监控 */
	pgstat_report_activity(STATE_RUNNING, debug_query_string);

	/* 附加到动态共享内存区。 */
	fc_area_space = shm_toc_lookup(fc_toc, PARALLEL_KEY_DSA, false);
	fc_area = dsa_attach_in_place(fc_area_space, fc_seg);

	/* 启动执行器 */
	fc_queryDesc->plannedstmt->jitFlags = fc_fpes->jit_flags;
	ExecutorStart(fc_queryDesc, fc_fpes->eflags);

	/* 并行工作进程的特殊执行器初始化步骤 */
	fc_queryDesc->planstate->state->es_query_dsa = fc_area;
	if (DsaPointerIsValid(fc_fpes->param_exec))
	{
		char	   *fc_paramexec_space;

		fc_paramexec_space = dsa_get_address(fc_area, fc_fpes->param_exec);
		fc_RestoreParamExecParams(fc_paramexec_space, fc_queryDesc->estate);
	}
	fc_pwcxt.toc = fc_toc;
	fc_pwcxt.seg = fc_seg;
	fc_ExecParallelInitializeWorker(fc_queryDesc->planstate, &fc_pwcxt);

	/* 传递任何元组限制 */
	ExecSetTupleBound(fc_fpes->tuples_needed, fc_queryDesc->planstate);

	/*
	 * 准备在查询执行期间跟踪缓冲区/WAL 使用情况。
	 *
	 * 我们在启动执行器之后执行此操作，以匹配领导者中的情况，
	 * 领导者在启动执行器期间也不计算缓冲区访问和 WAL 活动。
	 */
	InstrStartParallelQuery();

	/*
	 * 运行计划。如果我们指定了元组限制，请小心不要请求
	 * 超过该限制的元组。
	 */
	ExecutorRun(fc_queryDesc,
				ForwardScanDirection,
				fc_fpes->tuples_needed < 0 ? (int64) 0 : fc_fpes->tuples_needed,
				true);

	/* 关闭执行器 */
	ExecutorFinish(fc_queryDesc);

	/* 报告并行执行期间的缓冲区/WAL 使用情况。 */
	fc_buffer_usage = shm_toc_lookup(fc_toc, PARALLEL_KEY_BUFFER_USAGE, false);
	fc_wal_usage = shm_toc_lookup(fc_toc, PARALLEL_KEY_WAL_USAGE, false);
	InstrEndParallelQuery(&fc_buffer_usage[ParallelWorkerNumber],
						  &fc_wal_usage[ParallelWorkerNumber]);

	/* 如果设置了任何仪器选项，则报告仪器数据。 */
	if (fc_instrumentation != NULL)
		fc_ExecParallelReportInstrumentation(fc_queryDesc->planstate,
										  fc_instrumentation);

	/* 如果有任何 JIT 仪器数据则报告 */
	if (fc_queryDesc->estate->es_jit && fc_jit_instrumentation != NULL)
	{
		Assert(ParallelWorkerNumber < fc_jit_instrumentation->num_workers);
		fc_jit_instrumentation->jit_instr[ParallelWorkerNumber] =
			fc_queryDesc->estate->es_jit->instr;
	}

	/* 必须在捕获仪器数据后执行此操作。 */
	ExecutorEnd(fc_queryDesc);

	/* 清理。 */
	dsa_detach(fc_area);
	FreeQueryDesc(fc_queryDesc);
	fc_receiver->rDestroy(fc_receiver);
}
